package com.wasp.collector.service.akka.cluster;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;

import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import com.wasp.collector.service.akka.StatsService;
import com.wasp.collector.utils.PropertiesUtil;

import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Address;
import akka.actor.Props;

/**
 * akka service
 * 
 * @author Mrfeng
 *
 */

public class AkkaService {

	private static ActorSystem system = null;
	private static ActorRef clusterClientActor = null;
	public static final String ROLE = "collectorSystem";
	private static final Object objLock = new Object();
	private static volatile boolean isInitial = false;
	public static void iniAkka(){
		if(isInitial){
			return;
		}
		synchronized (objLock) {
			if(!isInitial){
				if(PropertiesUtil.AkkaCluserServer()){
					String[] ports = PropertiesUtil.AkkaCluserServerPort();
					startupCluster(ports);
				}
				if(PropertiesUtil.AkkaCluserClient()){
					startupCluserClient();
				}
				//getSystem();
				isInitial = true;
			}
		}
	}
	
	public static ActorSystem getSystem() {
		if (system == null) {
			system = ActorSystem.create("MySystem", ConfigFactory.load());
		}
		return system;
	}

	public static void stopSystem() {
		if (system != null) {
			system.terminate();
		}
	}

	public static void startupCluster(String[] ports) {
		for (String port : ports) {
			// Override the configuration of the port
			Config config = ConfigFactory.parseString("akka.remote.netty.tcp.port=" + port)
					.withFallback(ConfigFactory.parseString("akka.cluster.roles = ["+ROLE+"]"))
					.withFallback(ConfigFactory.load("stats1"));

			ActorSystem system = ActorSystem.create("ClusterSystem", config);
			system.actorOf(Props.create(StatsService.class), "statsService");
		}
	}
	
	public static void startupCluserClient(){
	    ActorSystem system = ActorSystem.create("ClusterSystem", ConfigFactory.load("stats1"));
	    system.actorOf(Props.create(ClusterClient.class), "client");
	    
	}
	
	public static Address balance(){
		List<Address> nodesList = new ArrayList<>(ClusterClientHelper.getNodes());
		while(true){
			try {
				if(nodesList.isEmpty()){
					System.out.println("等等获取nodes");
					TimeUnit.SECONDS.sleep(5000);
				}
				nodesList = new ArrayList<>(ClusterClientHelper.getNodes());
				if(!nodesList.isEmpty()){
					break;
				}
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			
		}
		Address address = nodesList.get(ThreadLocalRandom.current().nextInt(nodesList.size()));
		return address;
	}
	
	public static String getServicePath(){
		return "/user/statsService";
	}
}
